#include "Network.h"
#include "utils/logger.h"
#include "OpOutput.h"
#include "OpBEGroupBy.h"

using namespace std;
using namespace internal;

OpOutput::OpOutput(const OutputOperation& operation) {
	group_by = new OpBEGroupBy(operation.best_effort_group_by());
	
	CHECK(group_by->getRowSize() >= sizeof(size_t) , "rowSize is shorter than sizeof(size_t)") ;
	rowSize = group_by->getRowSize() - sizeof(size_t) ;			// pomijamy hasha
	
	rowsPerPackage = BUFFERSIZE / rowSize;
	CHECK(rowsPerPackage > 0, "wrong test - buffer is smaller than rowSize");
	
	package.resize(network->nodes_count());
	next.resize(network->nodes_count());
	flags.resize(network->nodes_count());

	for(int i=0; i<network->nodes_count(); i++) {
		package[i] = new char[rowSize * rowsPerPackage];
		next[i] = package[i];
		flags[i] = GROUP_BY_MID_FLAG;
	}
	
	rows = group_by->getDoItResultRows() ;
}
	
int OpOutput::doit() {
	group_by->doit();
	
	last_row = group_by->getLast();
	
	//  tworzenie paczek i wysylanie
	LOG("tworze paczki i wysylam") ;
	
	int nr;
	while (!rows->empty()) {
		char* q = rows->back();
		while (q != last_row) {
			nr = *((size_t*) q) % network->nodes_count();
			memcpy(next[nr], q + sizeof(size_t) , rowSize);
			next[nr] += rowSize;
			if (next[nr] == package[nr] + rowsPerPackage * rowSize) send(nr);
			
			q += rowSize + sizeof(size_t) ;
		}
		delete [] (rows->back());
		rows->pop_back();
		if (!rows->empty()) last_row = rows->back() + (rowSize + sizeof(size_t))*TABLESIZE;
	}
	
	for (int i=0; i<network->nodes_count(); i++) {
		flags[i] = GROUP_BY_END_FLAG;
		send(i);
	}
	
	return -1;
}
  	
void OpOutput::send(int nr) {
	
	
	//	wypisywanie ! :
	//	< package[nr], next[nr] ]
	// zmienic na wysylanie !
	/*void *q = package[nr];
	while(q != next[nr]) {
		// wypisywanie specyficznych danych !
		LOG("%d %lf", *((int*)q), *((double*)(q+sizeof(int)))) ;
		q += rowSize;
	}*/
	///////////////////////////////////
	
	network->send(nr, package[nr], next[nr] - package[nr], flags[nr]);
	
	// reset current row pointer
	next[nr] = package[nr];
}

